/*
 * Copyright 2015 Fabian Hueske / Vasia Kalavri
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.xuele.learn.flink.book;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.StreamGroupedReduceOperator;

/**
 * Example program to demonstrate keyed transformation functions: keyBy, reduce.
 * @see StreamGroupedReduceOperator
 */
public class ReduceTest {
    public static final String TEMP = "thread：%s；value1：%s；value2：%s";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple2<String, Integer>> source = env.fromElements(Tuple2.of("fg", 1),
                Tuple2.of("fg", 2), Tuple2.of("hh", 12), Tuple2.of("hh", 1), Tuple2.of("sm", 10), Tuple2.of("sm", 3),
                Tuple2.of("fg", 5), Tuple2.of("hh", 15), Tuple2.of("sm", 11));
        source
                .keyBy(value -> value.f0)
                // reduce就是可以将之前滚动聚合的值和当前的值进行比较，value1是之前的值（前一个reduce方法产生的结果或第一个输入的值），value2是当前的值
                .reduce((value1, value2) -> {
                    System.out.println(String.format(TEMP, Thread.currentThread().getName(), value1, value2));
                    return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                })
                .print();
        env.execute();
    }
}
